package com.and.video.service.websocket;

import com.alibaba.fastjson.JSONObject;
import com.and.video.domain.Danmu;
import com.and.video.domain.constant.UserMomentsConstant;
import com.and.video.service.DanmuService;
import com.and.video.service.util.RocketMQUtil;
import com.and.video.service.util.TokenUtil;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

// 声明为Spring组件，并定义WebSocket服务端点
@Component
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    // 使用原子整数统计在线连接数，保证线程安全
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    // 线程安全的Map，存储所有WebSocket连接（键为sessionId，值为WebSocketService实例）
    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    // 当前WebSocket会话对象
    private Session session;

    // 当前会话ID
    private String sessionId;

    // 用户ID（通过token解析获得）
    private Long userId;

    // Spring应用上下文（用于获取Bean）
    private static ApplicationContext APPLICATION_CONTEXT;

    // 设置应用上下文（由Spring容器初始化时调用）
    public static void setApplicationContext(ApplicationContext applicationContext){
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }

    // WebSocket连接建立时调用
    @OnOpen
    public void openConnection(Session session, @PathParam("token") String token){
        try{
            // 验证token并获取用户ID
            this.userId = TokenUtil.verifyToken(token);
        }catch (Exception ignored){}

        this.sessionId = session.getId();
        this.session = session;

        // 管理连接映射
        if(WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId, this);
        }else{
            WEBSOCKET_MAP.put(sessionId, this);
            ONLINE_COUNT.getAndIncrement(); // 在线人数+1
        }

        logger.info("用户连接成功：" + sessionId + "，当前在线人数为：" + ONLINE_COUNT.get());
        try{
            this.sendMessage("0"); // 发送初始消息
        }catch (Exception e){
            logger.error("连接异常");
        }
    }

    // WebSocket连接关闭时调用
    @OnClose
    public void closeConnection(){
        if(WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement(); // 在线人数-1
        }
        logger.info("用户退出：" + sessionId + "当前在线人数为：" + ONLINE_COUNT.get());
    }

    // 收到客户端消息时调用
    @OnMessage
    public void onMessage(String message){
        logger.info("用户信息：" + sessionId + "，报文：" + message);
        if(!StringUtil.isNullOrEmpty(message)){
            try{
                // 群发消息给所有连接
                for(Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()){
                    WebSocketService webSocketService = entry.getValue();
                    // 获取RocketMQ生产者Bean
                    DefaultMQProducer danmusProducer = (DefaultMQProducer)APPLICATION_CONTEXT.getBean("danmusProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message", message);
                    jsonObject.put("sessionId", webSocketService.getSessionId());
                    // 创建消息对象（指定主题和编码）
                    Message msg = new Message(UserMomentsConstant.TOPIC_DANMUS, jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    // 异步发送消息到RocketMQ
                    RocketMQUtil.asyncSendMsg(danmusProducer, msg);
                }

                // 如果用户已认证，保存弹幕到数据库和Redis
                if(this.userId != null){
                     Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                     danmu.setUserId(userId);
                     danmu.setCreateTime(new Date());
                     DanmuService danmuService = (DanmuService)APPLICATION_CONTEXT.getBean("danmuService");
                     danmuService.asyncAddDanmu(danmu);
                     danmuService.addDanmusToRedis(danmu);
                }
            }catch (Exception e){
                logger.error("弹幕接收出现问题");
                e.printStackTrace();
            }
        }
    }

    // 发生错误时调用
    @OnError
    public void onError(Throwable error){
        // 错误处理（当前为空实现）
    }

    // 向客户端发送消息
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    // 定时任务：每5秒向所有客户端发送当前在线人数
    @Scheduled(fixedRate=5000)
    private void noticeOnlineCount() throws IOException {
        for(Map.Entry<String, WebSocketService> entry : WebSocketService.WEBSOCKET_MAP.entrySet()){
            WebSocketService webSocketService = entry.getValue();
            if(webSocketService.session.isOpen()){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount", ONLINE_COUNT.get());
                jsonObject.put("msg", "当前在线人数为" + ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }

    // Getter方法
    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }
}